package com.atguigu.sink

import com.atguigu.api.SensorReading
import org.apache.flink.streaming.api.scala._
import org.apache.flink.streaming.connectors.redis.RedisSink
import org.apache.flink.streaming.connectors.redis.common.config.FlinkJedisPoolConfig
import org.apache.flink.streaming.connectors.redis.common.mapper.{RedisCommand, RedisCommandDescription, RedisMapper}

/**
 * @ClassName RedisSinkTest
 * @Description
 * @Author Mr Yang
 * @Date 2020/8/31 23:31
 * @Version 1.0
 */
object RedisSinkTest {
  def main(args: Array[String]): Unit = {
    //获取执行环境
    val env = StreamExecutionEnvironment.getExecutionEnvironment
    //设置全局并行度
    env.setParallelism(1)

    //source
    val inputStream = env.readTextFile("F:\\work\\FlinkTutorial\\src\\main\\resources\\sensor.txt")
    //transform
    val dataStream = inputStream.map(
      data => {
        val dataArray = data.split(",")
        SensorReading(dataArray(0).trim, dataArray(1).trim.toLong, dataArray(2).trim.toDouble)
      }
    )

    val conf = new FlinkJedisPoolConfig.Builder()
      .setHost("localhost")
      .setPort(6379)
      .build()
    //sink
    dataStream.addSink( new RedisSink(conf, new MyRedisMapper()) )

    dataStream.print()
    env.execute("redis sink test")
  }
}


class MyRedisMapper() extends RedisMapper[SensorReading]{
  //定义保存到redis的命令
  override def getCommandDescription: RedisCommandDescription = {
    //把传感器id和温度值保存成哈希表 HSET key field value
    new RedisCommandDescription( RedisCommand.HSET, "sensor_temperature" )
  }
  //定义保存到redis的key（field）
  override def getKeyFromData(t: SensorReading): String =  t.id
  //定义保存到redis的value
  override def getValueFromData(t: SensorReading): String = t.temperature.toString
}